package com.upcoming.data.storm.reliablesend;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import java.util.Map;
import java.util.Random;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-17 13:26
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class MyBolt implements IRichBolt {

    private OutputCollector collector;

    private TopologyContext context;



    public MyBolt() {
        MyUtil.outLog2NC(this, "new MyBolt()");;
    }

    //初始化
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        MyUtil.outLog2NC(this, "Bolt2.prepare()");
        this.collector = outputCollector;
        this.context = topologyContext;
    }

    /**
     * 每个元组都需要执行
     * @param tuple
     */
    public void execute(Tuple tuple) {
        String msg = tuple.getString(0);
        if (new Random().nextBoolean()) {
            this.collector.ack(tuple);
            MyUtil.outLog2NC(this, msg + ":ack");
        } else {
            MyUtil.outLog2NC(this, msg + ":fail");
            this.collector.fail(tuple);
        }
    }

    public void cleanup() {
        MyUtil.outLog2NC(this, "Bolt2.cleanup()");
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
